/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package remoting.channel;

import common.exception.ExecutionException;
import common.exception.RemotingException;
import common.model.MagicUrl;
import common.model.Request;
import common.model.Response;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import remoting.channel.ChannelEventRunnable.ChannelState;
import remoting.store.DataStore;
import threadpools.ThreadPool;

public class AllChannelHandler extends WrappedChannelHandler {

  public AllChannelHandler(ChannelHandler handler, MagicUrl url, ThreadPool threadPool) {
    super(handler, url, threadPool);
  }

  @Override
  public void connected(Channel channel) throws RemotingException {
    ExecutorService executor = getExecutorService();
    try {
      executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
    } catch (Throwable t) {
      throw new ExecutionException("connect event", channel,
          getClass() + " error when process connected event .", t);
    }
  }

  @Override
  public void disconnected(Channel channel) throws RemotingException {
    ExecutorService executor = getExecutorService();
    try {
      executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED));
    } catch (Throwable t) {
      throw new ExecutionException("disconnect event", channel,
          getClass() + " error when process disconnected event .", t);
    }
  }

  @Override
  public void received(Channel channel, Object message) throws RemotingException {
    ExecutorService executor = getExecutorService();
    try {
      executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
    } catch (Throwable t) {
      //TODO A temporary solution to the problem that the exception information can not be sent to the opposite end after the thread pool is full. Need a refactoring
      //fix The thread pool is full, refuses to call, does not return, and causes the consumer to wait for time out
      if (message instanceof Request && t instanceof RejectedExecutionException) {
        Request request = (Request) message;
        if (request.isTwoWay()) {
          String msg = "Server side(" + url.getIp() + "," + url.getPort()
              + ") threadpool is exhausted ,detail msg:" + t.getMessage();
          Response response = new Response(request.getId(), request.getVersion());
          response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);
          response.setErrorMessage(msg);
          channel.sendMessage(response);
          return;
        }
      }
      throw new ExecutionException(message, channel,
          getClass() + " error when process received event .", t);
    }
  }

  @Override
  public void caught(Channel channel, Throwable exception) throws RemotingException {
    ExecutorService executor = getExecutorService();
    try {
      executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CAUGHT, exception));
    } catch (Throwable t) {
      throw new ExecutionException("caught event", channel,
          getClass() + " error when process caught event .", t);
    }
  }
}
